
#include <iostream>
#include <sstream>
#include <stdexcept>

#include "bocat/common/base64.hpp"
#include "bocat/common/logging.hpp"
#include "fekcab/packet.pb.h"

#include "store_service.hpp"


namespace	fekcab	{
namespace	server	{

///////////////////////////////////////////////////////////////////////
//

StoreService::StoreService( const std::string ini_filename, const std::string ini_section )
	:	_ini_filename( ini_filename ),
        _ini_section( ini_section ),
		_store_manager( ini_filename.c_str(), ini_section.c_str() ),
		_store_p( NULL )
{
	GOOGLE_PROTOBUF_VERIFY_VERSION;

	_store_p = _store_manager.getStore();
}

StoreService::~StoreService()
{
	google::protobuf::ShutdownProtobufLibrary();

	_store_p = NULL;
}


///////////////////////////////////////////////////////////////////////
//

int
StoreService::handle( const std::string& request_encoded, std::string& response_encoded )
{
	Packet	pkt_request;
	Packet	pkt_response;

	bool	with_response = false;

	std::string	request;
	try {
		request = bocat::common::base64_decode< std::string >( request_encoded );
	}
	catch( std::exception& e ) {
		std::ostringstream	msg;
		msg << "fail to decode base64 '" << request_encoded << "' -- " << e.what();
		//throw std::runtime_error( msg.str() );
		// just ignore it
		bocat::common::Logger*	logger_p = bocat::common::getLogger();
		logger_p->error( __LINE__, __FILE__, "%s", msg.str().c_str() );
		
		return	0;
	}

	if ( !pkt_request.ParseFromString( request ) ) {
		bocat::common::Logger*	logger_p = bocat::common::getLogger();
		logger_p->error( __LINE__, __FILE__, "fail to parse request protocol buffer -- %s", request.c_str() );

		//throw std::runtime_error("Fail to parse request protocol buffer.");
		// just ignore it
		return	0;
	}

	switch ( pkt_request.op() ) {
	
	case	Packet::PUT :
	case	Packet::MPUT :
		handlePUT( pkt_request, pkt_response );
		with_response = true;
		break;

	case	Packet::APPEND :
		handleAPPEND( pkt_request, pkt_response );
		with_response = true;
		break;

	case	Packet::MAPPEND :
		handleMAPPEND( pkt_request, pkt_response );
		with_response = true;
		break;

	case	Packet::PUTIGNORE :
	case	Packet::MPUTIGNORE :
		handlePUTIGNORE( pkt_request, pkt_response );
		with_response = true;
		break;

	case	Packet::GET :
		handleGET( pkt_request, pkt_response );
		with_response = true;
		break;

	case	Packet::INC :
	case	Packet::DEC :
		handleINCorDEC( pkt_request, pkt_response );
		with_response = true;
		break;

	case	Packet::DELETE :
	case	Packet::MDELETE :
		handleDELETE( pkt_request, pkt_response );
		with_response = true;
		break;

	case	Packet::PING	:
		pkt_response = pkt_request;

		pkt_response.set_type( Packet::RESPONSE );
		pkt_response.set_reqid( pkt_request.sid() );
		pkt_response.set_sid( 1 );

		with_response = true;

		break;

	default	:
		std::ostringstream	msg;
		msg << "unsupport service OP : " <<  pkt_request.op();
		throw std::runtime_error( msg.str() );
		break;
	}

	if ( with_response ) {
		std::string	response;
		if ( !pkt_response.SerializeToString( &response ) ) {
			std::ostringstream	msg;
			msg << "fail to serialize response to string.";
			throw std::runtime_error( msg.str() );
		}

		response_encoded = bocat::common::base64_encode( response );
		return 1;
	}

	return	0;
}


///////////////////////////////////////////////////////////////////////
//

/*
static  u_int32_t
getPIN( const char* symbol, u_int32_t symbol_size )
{
    u_int32_t   pin = 0;
	if (symbol_size < 1) {
		return pin;
	}

	if ( symbol_size >= (2 + sizeof(u_int32_t)) ) {
		memcpy( (char *)&pin, &symbol[2], sizeof(u_int32_t) );
	}
	else if ( symbol_size == sizeof(u_int32_t) ) {
		memcpy( (char *)&pin, symbol, sizeof(u_int32_t) );
	}
	else {
		char*	p = (char *)&pin;
		for (u_int32_t	i=0, k=0; i < sizeof(u_int32_t); ++i ) {
			p[i] = symbol[ k ];

			++k;
			if (k >= symbol_size) {
				k = 0;
			}
		}
		
	}

    return pin;
}
*/

int
StoreService::handlePUT( Packet& pkt_request, Packet& pkt_response )
{
	pkt_response.set_type( Packet::RESPONSE );
	pkt_response.set_reqid( pkt_request.sid() );
	pkt_response.set_sid( 1 );
	pkt_response.set_op( pkt_request.op() );

	for( u_int32_t i=0; i < pkt_request.item_size(); i++ ) {
		Packet::Item*	item_p = pkt_response.add_item();

		const Packet::Item& item = pkt_request.item( i );
		std::string	 	table = item.table();
		std::string	key = item.key();
		std::string	data = item.value();
		std::string	kpin = item.kpin();

		int			rc,status = 200;
		try {
			if ( table.size() > 0 ) {
				rc = _store_p->insert( table, kpin, key, data );
				item_p->set_table( table );
			}
			else {
				rc = _store_p->insert( kpin, key, data );
			}
            if ( rc != 0 ) {
                status = 400;
            }

            item_p->set_kpin( kpin );

			item_p->set_key( key );
			item_p->set_status( status );
		}
		catch(std::exception& e) {
			std::ostringstream	msg;
			msg << "fail to PUT(key:" << key << ") -- " << e.what();

			status = 500;

			item_p->set_key( key );
			item_p->set_status( status );
			item_p->set_value( msg.str() );
		}
	}

	return	0;
}	//// Store::Service::handlePUT()



int
StoreService::handleAPPEND( Packet& pkt_request, Packet& pkt_response )
{
	pkt_response.set_type( Packet::RESPONSE );
	pkt_response.set_reqid( pkt_request.sid() );
	pkt_response.set_sid( 1 );
	pkt_response.set_op( pkt_request.op() );

	for( u_int32_t i=0; i < pkt_request.item_size(); i++ ) {
		Packet::Item*	item_p = pkt_response.add_item();

		const Packet::Item& item = pkt_request.item( i );
		std::string	 	table = item.table();
		std::string	key = item.key();
		std::string	data = item.value();
		std::string	kpin = item.kpin();

		int			rc,status = 200;
		try {
			if ( table.size() > 0 ) {
				rc = _store_p->insertAppend( table, kpin, key, data );
				item_p->set_table( table );	
			}
			else {
				rc = _store_p->insertAppend( kpin, key, data );
			}
            if ( rc != 0 ) {
                status = 400;
            }

            item_p->set_kpin( kpin );

			item_p->set_key( key );
			item_p->set_status( status );
		}
		catch(std::exception& e) {
			std::ostringstream	msg;
			msg << "fail to APPEND(key:" << key << ") -- " << e.what();

			status = 500;

			item_p->set_key( key );
			item_p->set_status( status );
			item_p->set_value( msg.str() );
		}
	}

	return	0;
}	//// Store::Service::handleAPPEND()



int
StoreService::handlePUTIGNORE( Packet& pkt_request, Packet& pkt_response )
{
	pkt_response.set_type( Packet::RESPONSE );
	pkt_response.set_reqid( pkt_request.sid() );
	pkt_response.set_sid( 1 );
	pkt_response.set_op( pkt_request.op() );

	for( u_int32_t i=0; i < pkt_request.item_size(); i++ ) {
		Packet::Item*	item_p = pkt_response.add_item();

		const Packet::Item& item = pkt_request.item( i );
		std::string	 	table = item.table();
		std::string	key = item.key();
		std::string	data = item.value();
		std::string	kpin = item.kpin();

		int			rc,status = 200;
		try {
			if ( table.size() > 0 ) {
				rc = _store_p->insertIgnore( table, kpin, key, data );
				item_p->set_table( table );
			}
			else {
				rc = _store_p->insertIgnore( kpin, key, data );
			}
            if ( rc != 0 ) {
                status = 400;
            }

            item_p->set_kpin( kpin );

			item_p->set_key( key );
			item_p->set_status( status );
		}
		catch(std::exception& e) {
			std::ostringstream	msg;
			msg << "fail to PUTIGNORE(key:" << key << ") -- " << e.what();

			status = 500;

			item_p->set_key( key );
			item_p->set_status( status );
			item_p->set_value( msg.str() );
		}
	}

	return	0;
}	//// Store::Service::handlePUTIGNORE()



///////////////////////////////////////////////////////////////////////
//


int
StoreService::handleMAPPEND( Packet& pkt_request, Packet& pkt_response )
{
	pkt_response.set_type( Packet::RESPONSE );
	pkt_response.set_reqid( pkt_request.sid() );
	pkt_response.set_sid( 1 );
	pkt_response.set_op( pkt_request.op() );

	std::vector< std::string >	kpin_v, key_v, data_v, table_v;
	size_t	count_valid_table = 0;

	for( u_int32_t i=0; i < pkt_request.item_size(); i++ ) {
		const Packet::Item& item = pkt_request.item( i );
		std::string	 	table = item.table();
		std::string	key = item.key();
		std::string	data = item.value();
		std::string	kpin = item.kpin();
		
		if ( table.size() > 0 ) {
			++count_valid_table;
		}

		table_v.push_back( table );
		kpin_v.push_back( kpin );
		key_v.push_back( key );
		data_v.push_back( data );
	}

	int			rc,status = 200;
	try {
		if ( count_valid_table > 0 ) {
			rc = _store_p->insertAppend( table_v, kpin_v, key_v, data_v );
		}
		else {
			rc = _store_p->insertAppend( kpin_v, key_v, data_v );
		}
		if ( rc != 0 ) {
			status = 400;
		}

		for (size_t k=0; k<kpin_v.size(); ++k) {
			Packet::Item*	item_p = pkt_response.add_item();

			const std::string&	kpin = kpin_v[ k ];
			const std::string&	key  = key_v[ k ];
			item_p->set_kpin( kpin );
			item_p->set_key( key );
			item_p->set_status( status );
		}
	}
	catch(std::exception& e) {
		std::ostringstream	msg;
		msg << "fail to Bulk-APPEND -- " << e.what();

		status = 500;

		for (size_t k=0; k<kpin_v.size(); ++k) {
			Packet::Item*	item_p = pkt_response.add_item();

			const std::string&	kpin = kpin_v[ k ];
			const std::string&	key  = key_v[ k ];

			item_p->set_kpin( kpin );
			item_p->set_key( key );
			item_p->set_status( status );
			item_p->set_value( msg.str() );
		}
	}


	return	0;
}	//// Store::Service::handleAPPEND()



///////////////////////////////////////////////////////////////////////
//


int
StoreService::handleGET( Packet& pkt_request, Packet& pkt_response )
{
	pkt_response.set_type( Packet::RESPONSE );
	pkt_response.set_reqid( pkt_request.sid() );
	pkt_response.set_sid( 1 );
	pkt_response.set_op( pkt_request.op() );

	int		limits = pkt_request.limit();

	for( u_int32_t i=0; i < pkt_request.item_size(); i++ ) {
		const Packet::Item& item = pkt_request.item( i );
		std::string	 	table = item.table();
		std::string	key = item.key();
		std::string	data;

		std::vector< std::string >	buf_v;
		int		rc, status = 200;
		try {
			std::string	kpin = item.kpin();

			if ( table.size() > 0 ) {
				rc = _store_p->find( table, kpin, key, buf_v, limits );
			}
			else {
				rc = _store_p->find( kpin, key, buf_v, limits );
			}
            if ( rc == 0 ) {
                status = 100;
            }
            else if ( rc < 0 ) {
                status = 400;
            }

			if ( rc > 0 ) {
				for (size_t k=0; k < buf_v.size(); ++k) {
					Packet::Item*	item_p = pkt_response.add_item();

					const std::string&	buf = buf_v[ k ];

					if ( table.size() > 0 ) {
						item_p->set_table( table );
					}
					item_p->set_kpin( kpin );
					item_p->set_key( key );
					item_p->set_value( buf );
					item_p->set_status( 200 );
				}
			}
			else {
				Packet::Item*	item_p = pkt_response.add_item();

				if ( table.size() > 0 ) {
					item_p->set_table( table );
				}
				item_p->set_kpin( kpin );
				item_p->set_key( key );
				item_p->set_status( status );
			}
		}
		catch(std::exception& e) {
			std::ostringstream	msg;
			msg << "fail to PUT(table:" << table << " key:" << key << ") -- " << e.what();

			status = 500;

			Packet::Item*	item_p = pkt_response.add_item();

			item_p->set_table( table );
			item_p->set_key( key );
			item_p->set_status( status );
			item_p->set_value( msg.str() );
		}
	}

	return	0;
}	//// Store::Service::handleGET()


///////////////////////////////////////////////////////////////////////
//

int
StoreService::handleDELETE( Packet& pkt_request, Packet& pkt_response )
{
	pkt_response.set_type( Packet::RESPONSE );
	pkt_response.set_reqid( pkt_request.sid() );
	pkt_response.set_sid( 1 );
	pkt_response.set_op( pkt_request.op() );

	for( u_int32_t i=0; i < pkt_request.item_size(); i++ ) {
		Packet::Item*	item_p = pkt_response.add_item();

		const Packet::Item& item = pkt_request.item( i );
		std::string	 	table = item.table();
		std::string	key = item.key();

		int			rc,status = 200;
		try {
            std::string	kpin = item.kpin();

			if ( table.size() > 0 ) {
				rc = _store_p->erase( table, kpin, key );
			}
			else {
				rc = _store_p->erase( kpin, key );
			}
            if ( rc != 0 ) {
                status = 400;
            }

            item_p->set_kpin( kpin );

			item_p->set_key( key );
			item_p->set_status( status );
		}
		catch(std::exception& e) {
			std::ostringstream	msg;
			msg << "fail to PUT(table:" << table << " key:" << key << ") -- " << e.what();

			status = 500;

			item_p->set_table( table );
			item_p->set_key( key );
			item_p->set_status( status );
			item_p->set_value( msg.str() );
		}
	}

	return	0;
}	//// Store::Service::handleDELETE()


///////////////////////////////////////////////////////////////////////
//

int
StoreService::handleINCorDEC( Packet& pkt_request, Packet& pkt_response )
{
	pkt_response.set_type( Packet::RESPONSE );
	pkt_response.set_reqid( pkt_request.sid() );
	pkt_response.set_sid( 1 );
	pkt_response.set_op( pkt_request.op() );

	int		limits = 1;
 
	for( u_int32_t i=0; i < pkt_request.item_size(); i++ ) {
		Packet::Item*	item_p = pkt_response.add_item();

		const Packet::Item& item = pkt_request.item( i );
		std::string	table = item.table();
		std::string	key = item.key();
		std::string	data_new;

		std::vector< std::string >	buf_v;
		int		rc, status = 200;
		try {
			std::string	kpin = item.kpin();
			std::string	error;

			if ( table.size() > 0 ) {
				rc = _store_p->find( table, kpin, key, buf_v, limits );
			}
			else {
				rc = _store_p->find( kpin, key, buf_v, limits );
			}
            if ( rc == 0 ) {
                status = 100;
            }
            else if ( rc < 0 ) {
                status = 400;

				std::ostringstream	msg;
				msg << "fail to find data";
				error = msg.str();
            }

			if ( status <= 200 ) {
				int64_t	value_old = 0;
				if ( 200 == status ) {
					const std::string&  buf = buf_v[ 0 ];

					try {
						const std::string&	value_b64 = buf;
						std::string	value = bocat::common::base64_decode< std::string >( value_b64 );
						if ( sizeof(int64_t) == value.size() ) {
							memcpy( &value_old, value.c_str(), value.size() );
						}
						else {
							/// TODO ERROR : BAD format
						}
					}
					catch( ... ) {
						/// TODO ERROR : BAD format
					}
				}
				else {
					// set default status
					status = 200;
				}

				// operate
				std::string	delta_str = item.value();
				if ( sizeof(int64_t) != delta_str.size() ) {
					status = 300;

					std::ostringstream	msg;
					msg << "BAD format (size=" << delta_str.size() << ", " << sizeof(int64_t) << " wanted)";
					error = msg.str();
				}
				else {
					int64_t	value_new = 0;
					int64_t	delta;
					memcpy( &delta, delta_str.c_str(), delta_str.size() );
					if ( Packet::INC == pkt_request.op() ) {
						value_new = value_old + delta;
					}
					else {
						value_new = value_old - delta;
					}

					// update 
					data_new.assign( (char *)&value_new, sizeof(int64_t) );
					std::string	data_new_b64 = bocat::common::base64_encode( (char *)&value_new, sizeof(int64_t) );
					if ( table.size() > 0 ) {
						rc = _store_p->insert( table, kpin, key, data_new_b64 );
						item_p->set_table( table );
					}
					else {
						rc = _store_p->insert( kpin, key, data_new_b64 );
					}
					if ( rc != 0 ) {
						status = 400;

						std::ostringstream	msg;
						msg << "fail to update data";
						error = msg.str();
					}
				}
			}

			item_p->set_kpin( kpin );

			item_p->set_key( key );
			item_p->set_status( status );
			if ( status == 200 ) {
				item_p->set_value( data_new );
			}
			else if ( status > 200 ) {
				item_p->set_value ( error );
			}
		}
		catch(std::exception& e) {
			std::ostringstream	msg;
			if ( Packet::INC == pkt_request.op() ) {
				msg << "fail to INC(table:" << table << " key:" << key << ") -- " << e.what();
			}
			else {
				msg << "fail to DEC(table:" << table << " key:" << key << ") -- " << e.what();
			}

			status = 500;

			item_p->set_table( table );
			item_p->set_key( key );
			item_p->set_status( status );
			item_p->set_value( msg.str() );
		}
	}

	return	0;
}	//// Store::Service::handleINCorDEC()


///////////////////////////////////////////////////////////////////////
//

int
StoreService::slim( void )
{
	return	_store_p->slim();
}

///////////////////////////////////////////////////////////////////////
//


}	//// namespace	server
}	//// namespace	fekcab


